package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.handler.RpcClientHandler;
import lombok.SneakyThrows;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.lang.reflect.Proxy;
import java.security.SecureRandom;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 *
 * @author rephilo
 */
@Component
public class RpcClientProxy {

    public volatile List<RpcClient> rpcClientList;

    @Autowired
    private RpcClientHandler rpcClientHandler;

    private Random random;

    @Autowired
    private CuratorFramework client;

    private PathChildrenCache childrenCache;

    private ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(4);

    public Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, (proxy, method, args) -> {
                    //1.封装request请求对象
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setRequestId(UUID.randomUUID().toString());
                    rpcRequest.setClassName(serviceClass.getName());
                    rpcRequest.setMethodName(method.getName());
                    rpcRequest.setParameterTypes(method.getParameterTypes());
                    rpcRequest.setParameters(args);
                    //2.创建RpcClient对象
                    RpcClient rpcClient = getRandomClient();
                    try {
                        //3.发送消息
                        Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));

                        RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                        if (rpcResponse.getError() != null) {
                            throw new RuntimeException(rpcResponse.getError());
                        }
                        //4.返回结果
                        Object result = rpcResponse.getResult();
                        return JSON.parseObject(result.toString(), method.getReturnType());
                    } catch (Exception e) {
                        throw e;
                    }
                });
    }

    private void degradeNode(RpcClient rpcClient) throws Exception {
        String path = "/" + rpcClient.getIp() + ":" + rpcClient.getPort();
        client.setData().forPath(path, "0".getBytes());
    }

    private void recoverNode(RpcClient rpcClient) throws Exception {
        String path = "/" + rpcClient.getIp() + ":" + rpcClient.getPort();
        client.setData().forPath(path, "10".getBytes());
    }

    @PostConstruct
    public void initClient() throws Exception {
        this.random = new Random();
        initRpcList();
        //注册监听
        childrenCache = new PathChildrenCache(client, "/", true);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        childrenCache.getListenable().addListener((framework, event) -> {
            switch (event.getType()) {
                case CHILD_ADDED:
                case CHILD_UPDATED:
                case CHILD_REMOVED:
                    initRpcList();
                    break;
                default:
                    break;
            }
        });

        //健康检查线程池
        pool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    healthCheck(rpcClientList);
                } catch (Exception ignored) {

                }
            }
        }, 1, 5, TimeUnit.SECONDS);
    }

    /**
     * 健康检查过程
     *
     * @param rpcClientList
     */
    private void healthCheck(List<RpcClient> rpcClientList) throws Exception {
        for (RpcClient rpcClient : rpcClientList) {
            long start = System.currentTimeMillis();
            //模拟健康检查通讯
            long end = System.currentTimeMillis();
            if (end - start >= 5000) {
                //降级节点
                degradeNode(rpcClient);
            } else {
                //降级的节点恢复
                if (rpcClient.getWeight() == 0) {
                    recoverNode(rpcClient);
                }
            }
        }
    }

    private void initRpcList() {
        try {
            List<String> strings = client.getChildren()
                    .forPath("/");
            this.rpcClientList = strings.stream()
                    .map(this::getClient)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
        } catch (Exception ignored) {
        }
    }

    @PreDestroy
    public void closeClient() {
        for (RpcClient rpcClient : rpcClientList) {
            rpcClient.close();
        }
    }

    /**
     * 将zk的内容转换为实体
     *
     * @param nodeName
     * @return
     */
    private RpcClient getClient(String nodeName) {
        String[] split = nodeName.split(":");
        String host = split[0].replace("/", "");
        String port = split[1];
        try {
            byte[] bytes = client.getData().forPath("/" + nodeName);
            String weight = new String(bytes);
            return new RpcClient(host, Integer.parseInt(port), Integer.parseInt(weight), rpcClientHandler);
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 随机获取client的对象
     *
     * @return
     */
    private RpcClient getRandomClient() {
        List<RpcClient> collect = rpcClientList.stream()
                .filter(x -> x.getWeight() > 0)
                .collect(Collectors.toList());
        if (!collect.isEmpty()) {
            int n = this.random.nextInt(collect.size());
            return collect.get(n);
        }

        return null;
    }
}
